package spring.cloud.tasks.executor.name_space.executor;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.utils.CloseableExecutorService;
import spring.cloud.tasks.common.ExecutorServiceUtil;
import spring.cloud.tasks.common.ThreadFactory;
import spring.cloud.tasks.common.node.BlockNodePaths;
import spring.cloud.tasks.executor.name_space.NameSpaceDataManager;
import spring.cloud.tasks.executor.name_space.executor.task.task_listener.task_item_allot.TaskItemAllotContentService;
import spring.cloud.tasks.executor.name_space.executor.tasks_runtime_info.task_item_assign.TaskIdToTaskInstanceIdToTaskItemIndexListMapMap;
import spring.cloud.tasks.executor.name_space.executor.tasks_runtime_info.task_item_assign.TaskItemAssignListener;
import spring.cloud.tasks.executor.utils.LogUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executors;

@Slf4j
@Setter
@Getter
public class TaskInitService {

    public NameSpaceDataManager nameSpaceDataManager;
    public String executorId;
    //
    public NodeCache nodeCache;
    //
    public TreeCache treeCache;
    public java.util.concurrent.ExecutorService executorService;
    public List<String> taskIdSet = new ArrayList<>();
    //
    public TaskItemAllotContentService taskItemAllotContentService;
    public TaskIdToTaskInstanceIdToTaskItemIndexListMapMap taskIdToTaskInstanceIdToTaskItemIndexListMapMap;

    public TaskInitService(ExecutorInitAndTaskInitService executorInitAndTaskInitService) {
        this.nameSpaceDataManager = executorInitAndTaskInitService.getNameSpaceDataManager();
        this.executorId = executorInitAndTaskInitService.getExecutorId();
        this.taskItemAllotContentService = new TaskItemAllotContentService(executorInitAndTaskInitService.getNameSpaceDataManager().getCuratorFramework());
        this.taskIdToTaskInstanceIdToTaskItemIndexListMapMap = new TaskIdToTaskInstanceIdToTaskItemIndexListMapMap(executorId, taskItemAllotContentService);
    }

    public boolean removeTaskId(String taskId) {
        return taskIdSet.remove(taskId);
    }

    public void start() throws Exception {

        nodeCache = new NodeCache(nameSpaceDataManager.getCuratorFramework(), "/TaskItemAllot");
        nodeCache.start(true);
        //添加节点监听事件,NodeCacheListener每次都会触发，但不能获取监听的操作类型到底是添加还是删除等。
        nodeCache.getListenable().addListener(new TaskItemAssignListener(nodeCache, taskIdToTaskInstanceIdToTaskItemIndexListMapMap));
        //
        String enableTasksRootPath = BlockNodePaths.ENABLE_TASKS_NODE_PATH;
        treeCache = TreeCache.newBuilder(nameSpaceDataManager.getCuratorFramework(), enableTasksRootPath).setExecutor(new CloseableExecutorService(Executors.newSingleThreadExecutor(new ThreadFactory(executorId + "-$Jobs-watcher", false)), true)).setMaxDepth(1).build();
        executorService = Executors.newSingleThreadExecutor(new ThreadFactory(executorId + "-initNewJob-thread", false));
        treeCache.getListenable().addListener(new TaskInitListener(this), executorService);
        treeCache.start();
    }

    public void shutdown() {
        try {
            if (nodeCache != null) {
                nodeCache.close();
            }
        } catch (Throwable t) {
            LogUtils.error(log, "", t.toString(), t);
        }
        try {
            if (treeCache != null) {
                treeCache.close();
            }
        } catch (Throwable t) {
            LogUtils.error(log, "", t.toString(), t);
        }
        try {
            ExecutorServiceUtil.shutdownNow(executorService);
        } catch (Throwable t) {
            LogUtils.error(log, "", t.toString(), t);
        }
    }


}
